package com.JadePenG.Storm;

import org.apache.storm.topology.BasicOutputCollector;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseBasicBolt;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;

/**
 * 接收kafkaSpout发送的日志数据, 对数据进行处理,得到用户搜索的信息
 * <p>
 * BaseBasicBolt 与 BaseRichBolt 差别 少一个方法, 也不需要在初始化OutputCollector [BasicOutputCollector]
 */
public class SplitBolt extends BaseBasicBolt {

    /**
     * 接收kafkaSpout发送的日志数据 对数据进行处理,得到用户搜索的信息
     *
     * @param tuple
     * @param basicOutputCollector
     */
    @Override
    public void execute(Tuple tuple, BasicOutputCollector basicOutputCollector) {
        //1.获取上游kafkaSpout发送的日志数据
        String log = (String) tuple.getValue(4);//获取第5个元素
        //2.处理数据  #CS#
        if (log != null && log.contains("#CS#")) {//log不是空，并且字符串中包含#CS#
            int index = log.lastIndexOf("#CS#");
            String keywords = log.substring(index + 4);
            //将数据发送给下游
            basicOutputCollector.emit(new Values(keywords));
        }
    }

    /**
     * 声明定义
     *
     * @param outputFieldsDeclarer
     */
    @Override
    public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
        //起一个名字
        outputFieldsDeclarer.declare(new Fields("keywords"));
    }
}
